package com.example.messagedemo.producer;

import com.example.messagedemo.request.SendTopicMessageRequest;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.stereotype.Component;

import java.util.Date;

@Component
public class MessageProducer {

    final private DefaultMQProducer producer;

    public MessageProducer(DefaultMQProducer producer) {
        this.producer = producer;
    }


    /**
     * 发送一条同步消息
     * @param request
     * @throws Exception
     */
    public void sendCommonMessage(SendTopicMessageRequest request) throws Exception {
        byte[] messageByte = request.getMessage().getBytes(RemotingHelper.DEFAULT_CHARSET);
        Message msg = new Message(request.getTopic(), messageByte);
        // 设置消息标签,可选,默认为空
        msg.setTags("tag1");
        System.out.println((new Date()).toString());
        producer.send(msg, 10000);
    }


    /**
     * 发送一条同步且带消息轨迹的消息
     *
     * @param request
     */
    public void sendTraceMessage(SendTopicMessageRequest request) throws Exception {
        byte[] messageByte = request.getMessage().getBytes(RemotingHelper.DEFAULT_CHARSET);
        Message msg = new Message(request.getTopic(), messageByte);
        // 轨迹ID,从消息中心运维获取
        msg.setTraceId(request.getTraceId());
        // 轨迹进程ID, 如果为轨迹的开始主题,则需要自定义该ID,一般为UUID,根据实际业务需求定义
        // 如果为轨迹的中间节点需要传父消息ID和父消息回传的轨迹进程ID
        msg.setProcessId(request.getProcessId());
        // 父消息ID,多个使用英文逗号隔开
        msg.setParentId(request.getParentId());
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);
    }


    /**
     * 发送一条异步且带轨迹的消息
     *
     * @param request
     */
    public void sendSyncMessage(SendTopicMessageRequest request) throws Exception {
        Message msg = new Message(request.getTopic(), request.getMessage().getBytes(RemotingHelper.DEFAULT_CHARSET));
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("%s%n", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                System.out.printf("%s%n", e);
            }
        });

    }

    /***
     * 直接发送一条异步消息,不需要回调
     * @param request
     */
    public void sendOneWay(SendTopicMessageRequest request) throws Exception {
        Message msg = new Message(request.getTopic(), request.getMessage().getBytes(RemotingHelper.DEFAULT_CHARSET));
        msg.setTraceId(request.getTraceId());
        msg.setProcessId(request.getProcessId());
        msg.setParentId(request.getParentId());
        producer.sendOneway(msg);
    }
}
